package com.mavenq.fly.business.service;

import com.mavenq.fly.business.entity.TempData;
import com.mavenq.fly.business.mapper.TempDataMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
@Service
public class DataProducerService {

    private final AtomicBoolean running = new AtomicBoolean(false);

    @Value("${app.producer.fetch_size:100}")
    private int FETCH_SIZE;

    @Value("${app.producer.poll-interval:5}")
    private long POLL_INTERVAL_SECONDS;

    @Autowired
    private TempDataMapper tempDataMapper;

    @Autowired
    private BlockingQueue<TempData> dataQueue;

    private Thread producerThread;

    public synchronized void start() {
        if (running.get()) {
            log.warn("Producer is already running");
            return;
        }

        running.set(true);
        producerThread = new Thread(this::produceData, "Data-Producer");
        producerThread.start();
        log.info("Data producer started");
    }

    public synchronized void stop() {
        running.set(false);
        if (producerThread != null) {
            producerThread.interrupt();
        }
        log.info("Data producer stopped");
    }

    private void produceData() {
        while (running.get()) {
            try {
                List<TempData> dataList = tempDataMapper.selectPendingData(FETCH_SIZE);
                if (dataList.isEmpty()) {
                    log.debug("没有待处理数据，等待{}秒...", POLL_INTERVAL_SECONDS);
                    TimeUnit.SECONDS.sleep(POLL_INTERVAL_SECONDS);
                    continue;
                }

                for (TempData data : dataList) {
                    if (!running.get()) break;

                    if (markAsProcessing(data.getId())) {
                        boolean offered = dataQueue.offer(data, 1, TimeUnit.SECONDS);
                        if (!offered) {
                            tempDataMapper.updateStatus(data.getId(), TempData.Status.PENDING.name());
                            log.warn("队列已满，等待重试...");
                            TimeUnit.SECONDS.sleep(1);
                        }
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                log.info("生产者线程被中断");
                break;
            } catch (Exception e) {
                log.error("生产者错误: {}", e.getMessage(), e);
                try {
                    TimeUnit.SECONDS.sleep(10);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
        resetProcessingOnShutdown();
    }

    @Transactional
    public boolean markAsProcessing(String id) {
        int updated = tempDataMapper.updateStatus(id, TempData.Status.PROCESSING.name());
        return updated > 0;
    }

    public void resetProcessingOnShutdown() {
        try {
            int count = tempDataMapper.resetProcessingData();
            log.info("Reset {} PROCESSING records to PENDING on shutdown", count);
        } catch (Exception e) {
            log.error("Error resetting processing data: {}", e.getMessage(), e);
        }
    }
}